package middleware.kafka.Serializer;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 生产者拦截器
 *
 * @Author suxiaoxiao
 * @Date 2019/6/26 23:29
 * @Version 1.0
 */
public class ProducerInterceptorPrefix implements ProducerInterceptor<String, String> {

    /**
     * volatile 多线程共享值
     */
    private volatile long sendSuccess = 0;

    private volatile long sendFailure = 0;

    /**
     * KafkaProducer 在将消息序列化和计算分区之前会调用生产者拦截器的
     * onSend() 方法来对消息进行相应的定制化操作
     * @param producerRecord
     * @return
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        String modifiedValue = "prefix1-" + producerRecord.value();
        return new ProducerRecord<>(producerRecord.topic(),
                producerRecord.partition(), producerRecord.timestamp(),
                producerRecord.key(), modifiedValue, producerRecord.headers());
    }

    /**
     * KafkaProducer 会在消息被应答之前或消息发送失败时调用生产者拦截器的 onAcknowledgement() 方法，
     * 优先于用户设定的 Callback 之前执行。
     * 这个方法运行在 Producer 的I/O线程中，
     * 所以这个方法中实现的代码逻辑越简单越好，否则会影响消息的发送速度
     * @param recordMetadata
     * @param e
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (e == null) {
            sendSuccess++;
        } else {
            sendFailure++;
        }
    }

    /**
     * close() 方法主要用于在关闭拦截器时执行一些资源的清理工作
     * 这里我们统计了 kafka 发送的成功率
     */
    @Override
    public void close() {
        double successRatio = sendSuccess / (sendFailure + sendSuccess);
        System.out.println("[INFO] 发送成功率 = " + String.format("%f", successRatio * 100) + "%");
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}
